package com.lagou.bak;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos;

import java.sql.DriverManager;
import java.util.List;
import java.util.Map;

public class Test {
    public static void main(String[] args) throws Exception {


        DriverManager.getConnection("","","");
        /**
         * state
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> data = env.socketTextStream("hadoop2", 7777);
        SingleOutputStreamOperator<Tuple2<String, String>> maped = data.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String value) throws Exception {
                return new Tuple2<String, String>(value.split(",")[0], value.split(",")[1]);
            }
        });
        KeyedStream<Tuple2<String, String>, String> keyed = maped.keyBy(value -> value.f0);
        SingleOutputStreamOperator<Tuple2<String, String>> result = keyed.flatMap(new RichFlatMapFunction<Tuple2<String,String>, Tuple2<String, String>>() {
            ValueState<String> myState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> mystateDescripter = new ValueStateDescriptor<>("mystate", String.class);
                myState = getRuntimeContext().getState(mystateDescripter);
            }

            Integer sum = 0;

            @Override
            public void flatMap(Tuple2<String,String> value, Collector<Tuple2<String, String>> out) throws Exception {
                sum += Integer.valueOf(value.f1);
                myState.update(String.valueOf(sum));
                if (sum % 2 == 0) {
                    out.collect(new Tuple2<>(value.f0, String.valueOf(sum)));
                }
            }
        });
        result.print();
        env.execute();
    }
}
